package com.xf.day07_graphx
import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
 * 方式一 :
 *    使用 顶点RDD 和 边RDD 构建图
 */
object NewGraphx {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("testGraphx").setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 创建一个顶点集的 RDD,  VertexId 一个 Long 类型, 顶点属性是一个 二元组
    val users : RDD[( VertexId, (String, String))] = sc.parallelize(
      Array(
        ( 3L, ("rxin", "student")),
        ( 7L, ("jgonzal", "postdoc")),
        ( 5L, ("franklin", "prof")),
        ( 2L, ("istoica", "prof")))
    )

    // 创建一个边集的 RDD
    val relationships :RDD[ Edge[String]] = sc.parallelize(
      Array( Edge(3L ,7L, "collab") , Edge(5L ,3L, "advisor") , Edge(2L ,5L, "colleague") , Edge(5L ,7L, "pi") )
    )

    // 定义边中 用户缺失时 的 默认(缺失) 用户
    val defaultUser = ("John Doe", "Missing")

    // 使用 users 和 relationships 两个 RDD实例化 graph类 建立一个 Graph 对象
    val graph = Graph( users, relationships, defaultUser )

    // 查看图的 顶点信息
    /**
     * (2,(istoica,prof))
     * (3,(rxin,student))
     * (5,(franklin,prof))
     * (7,(jgonzal,postdoc))
     */
    // graph.vertices.collect().foreach(println)


    // 查看图的 边的信息
    /**
     * Edge(3,7,collab)
     * Edge(5,3,advisor)
     * Edge(2,5,colleague)
     * Edge(5,7,pi)
     */
    // graph.edges.collect().foreach(println)

    // 过滤出图上顶点的第二个属性值为 prof 的所有顶点   VertexRDD[(String, String)]  ==> 类型为啥是 VertexRDD[(String, String)]  ???
    val verticesCount :  VertexRDD[(String, String)] = graph.vertices.filter{ case (id, (name, pos)) => pos == "prof"}

    // 输出满足条件的 顶点个数
    val count = verticesCount.count()
    // println(count)  // 2

    // 输出满足条件的顶点  ==> (2,(istoica,prof))  和 (5,(franklin,prof))
    // verticesCount.collect().foreach( println)

    // 过滤出 原顶点ID 大于 目标顶点ID 的边
    val edgeCount :RDD[Edge[String]] = graph.edges.filter(e => e.srcId < e.dstId)
    val edgeCnt = edgeCount.count()
    // println(edgeCnt)   // 3

    /**
     * 具体的答应的结果为:
     *    Edge(3,7,collab)
     *    Edge(2,5,colleague)
     *    Edge(5,7,pi)
     */
    edgeCount.collect().foreach( println )
  }
}
